package org.jxkj.utils;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.jxkj.bean.PointData;

import java.io.IOException;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;
import java.util.TreeMap;

public class HBaseUtils {

    private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
    private static SimpleDateFormat hour = new SimpleDateFormat("yyyyMMddHH");

    /**
     * 根据pointname从redis获得pointid
     * @param pointName
     * @return
     */
    public static String[] getCodeAndId(String pointName){
        //从redis获得pointId ：001256
        String pointID = RedisUtils.get(pointName);
        int pid = 99;
        if(pointID==null || pointID.isEmpty())
        { pid = 999999;
        }else{
             pid = Integer.parseInt(pointID);
        }
        // int pid = Integer.valueOf(pointID.replaceAll("^(0+)", ""));
        //获得分区码
        int i = pid % 20;
        DecimalFormat df = new DecimalFormat("00");
        String regionCode = df.format(i);
        String[] result = {regionCode, String.format("%06d", pid)};
        return result;
    }


    /**
     * 针对高表
     * 查询返回数据
     * @param scan
     * @return
     */
    public static List<PointData> doScan(Table table,Scan scan) throws IOException {
        ResultScanner scanner = table.getScanner(scan);
        List<PointData> list = new ArrayList<>();
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                PointData data = new PointData();
                String row = Bytes.toString(CellUtil.cloneRow(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                if(!value.equals("")){
                    data.setValue(value);
                    data.setTime(row.split("\\|")[2]);
                    list.add(data);
                }
            }
        }
        return list;
    }


    /**
     * 针对宽表
     * 处理、封装数据方法
     * rowkey:00|015450|2019092000
     * @param scan
     * @return
     */
    /*
    public static List<PointData> genData(Table table,Scan scan,String pointName) throws Exception {
        List<PointData> list = new ArrayList<>();
        //得到 resultcanner 实现类的对象

        ResultScanner resultScanner = table.getScanner(scan);

        for (Result result : resultScanner) {
            //一个result对象里面封装的是一个rowkey的所有信息，在这里就可以获得rowKey
            String rowKey = Bytes.toString(result.getRow());
            String[] split = rowKey.split("\\|");
            //获得yyyy-MM-dd 00:00:00
            String startHour = sdf.format(hour.parse(split[2]));
            long startTimestamp = ParseUtils.getTimestamp(startHour);
            Cell[] cells = result.rawCells();
            //一个cells对象封装了一个rowkey所有的单元格
            for (Cell cell : cells) {
                PointData data = new PointData();
                //获得列所表示的秒
                Integer second = Integer.valueOf(Bytes.toString(CellUtil.cloneQualifier(cell)));
                long timeUTC = startTimestamp + second;
                data.setTime(ParseUtils.getDate(timeUTC));
                data.setValue(Bytes.toString(CellUtil.cloneValue(cell)));
                data.setName(pointName);
                data.setTimestamp(timeUTC);
                list.add(data);
            }
        }
        resultScanner.close();
        return list;

    }
*/
    public static List<PointData> genData(Table table, Scan scan, String pointName) throws Exception {
        TreeMap<Long, String> map = new TreeMap<>();
        //得到 resultcanner 实现类的对象
        ResultScanner resultScanner = table.getScanner(scan);
        for (Result result : resultScanner) {
            //一个result对象里面封装的是一个rowkey的所有信息，在这里就可以获得rowKey
            String rowKey = Bytes.toString(result.getRow());
            String[] split = rowKey.split("\\|");
            //获得yyyy-MM-dd 00:00:00
            String startHour = sdf.format(hour.parse(split[2]));
            long startTimestamp = ParseUtils.getTimestamp(startHour);
            Cell[] cells = result.rawCells();
            //一个cells对象封装了一个rowkey所有的单元格
            for (Cell cell : cells) {
                //获得列所表示的秒
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                if(!column.contains("x")){
                    Integer second = Integer.valueOf(Bytes.toString(CellUtil.cloneQualifier(cell)));
                    long timeUTC = startTimestamp + second;
                    map.put(timeUTC,Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
        }
        List<PointData> list = new ArrayList<>();
        for (Long timeUTC : map.keySet()) {
            PointData data = new PointData();
            data.setTime(ParseUtils.getDate(timeUTC));
            data.setName(pointName);
            data.setTimestamp(timeUTC);
            data.setValue(map.get(timeUTC));
            list.add(data);
        }
        return list;
    }

    /**
     * 读取result一个rowkey为一个排序map
     * key:时间戳 value：字符串值
     * @param rs 结果集
     * @return
     */
    public static NavigableMap<Long, String> readRowFamilyMap(Result rs) {
        NavigableMap<Long, String> map = new TreeMap<>();
        NavigableMap<byte[], byte[]> resultMap = rs.getFamilyMap(Bytes.toBytes("CF"));
        long ts = getTsByRowkey(Bytes.toString(rs.getRow()));
        resultMap.forEach((k, v) -> {
            if(!Bytes.toString(k).contains("x")) {
                int key = Integer.parseInt(Bytes.toString(k));
                String val = Bytes.toString(v);
                map.put(ts + key, val);
            }
        });
        return map;
    }
    /**
     * rowkey转换为初始时间戳
     * @param rowkey
     * @return
     */
    public static long getTsByRowkey(String rowkey) {
        String[] split = rowkey.split("\\|");
        //获得yyyy-MM-dd 00:00:00
        String hour = ParseUtils.dateToStr(ParseUtils.strToDate(split[2],"yyyyMMddHH"));
        long ts = ParseUtils.getTimestamp(hour);
        return ts;
    }
    /**
     * 组装pointData对象
     * @param map
     * @return
     */
    public static List<PointData> genBean(NavigableMap<Long,String> map,String pointName){
        List<PointData> list = new ArrayList<>();
        for (Long timeUTC : map.keySet()) {
            PointData data = new PointData();
            data.setTime(ParseUtils.getDate(timeUTC));
            data.setName(pointName);
            data.setTimestamp(timeUTC);
            data.setValue(map.get(timeUTC));
            list.add(data);
        }
        return list;
    }

    /**
     *
     * 生成分区码
     * pointID除以regions的余数
     * @param regions
     * @return
     */
    public static String genRegionCode(int pointID, int regions){
        int i = pointID % regions;
        DecimalFormat df = new DecimalFormat("00");
        String regionCode = df.format(i);
        return regionCode;
    }

    /**
     * 生成rowkey
     * regionCode|point_ID|timestamp
     * @return
     */
    public static String genRowKey(String regionCode, String pointID, String timestamp){
        StringBuilder sb = new StringBuilder();
        sb.append(regionCode + "|")
                .append(pointID + "|")
                .append(timestamp );
        return sb.toString();
    }


}
